REGISTER /home/geil/pig/kmgross/piggybank-0.12.0.jar;
DEFINE ISOToUnix org.apache.pig.piggybank.evaluation.datetime.convert.ISOToUnix();
rawdata = load 'cql://kmexample/rawdata2'
          using CqlStorage() as
           (firstname:tuple(name:chararray,value:chararray),
            surname:tuple(name:chararray,value:chararray),
            date:tuple(name:chararray,value:long),
            fromcity:tuple(name:chararray,value:chararray),
            km:tuple(name:chararray,value:int),
            tocity:tuple(name:chararray,value:chararray)
           );
flat = foreach rawdata generate
       firstname.value as firstname,
       surname.value as surname,
       date.value as date:long,
       km.value as km;
filt = filter flat by date > ISOToUnix('$startDate') 
                      and date < ISOToUnix('$endDate');
grouped = group filt by (firstname, surname);
agg = foreach grouped 
       generate group.firstname as firstname, 
                group.surname as surname, 
                SUM(filt.km) as km;
ord = order agg by km desc;
topX = limit ord $topX;
cassandra = foreach topX generate 
      TOTUPLE(TOTUPLE('firstname',firstname),
              TOTUPLE('surname',surname)),
			TOTUPLE(km);
store cassandra
    into 'cql://kmexample/topx?output_query=update+topx+set+km+%3D+%3F'
    using CqlStorage();